package cn.doitedu.kafka.demos;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Random;

public class ProducerDemo {

    public static void main(String[] args) throws InterruptedException {

        Properties props = new Properties();
        /**
         * 必配参数
         */
        //设置kafka集群的地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "doit01:9092,doit02:9092,doit03:9092");
        //序列化器
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


        /**
         * 可选参数
         */
        //ack模式，取值有0，1，-1（all）  ， all是最慢但最安全的
        //props.put("acks", "all");
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        //失败重试次数（有可能会造成数据的乱序）
        props.put("retries", 3);
        //数据发送的批次大小
        props.put("batch.size", 10);
        //生产者一次请求向服务器发送的数据量的最大字节数
        // props.put("max.request.size", 20);
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,204800);
        //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
        props.put("linger.ms", 10000);
        //整个Producer用到总内存的大小，如果缓冲区满了会提交数据到服务端
        //buffer.memory要大于batch.size，否则会报申请内存不足的错误
        props.put("buffer.memory", 102400000);  // 默认32M

        // 构造一个生产者实例对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        Random rd = new Random();
        for(int i=0;i<10000;i++){
            String data = i+ " kafka is greate,doit is more greater , " + rd.nextInt(10000);
            Thread.sleep(500+rd.nextInt(2000));
            producer.send(new ProducerRecord<String,String>("doit27-1",i+"",data));

        }

        producer.flush();
        producer.close();


    }

}
